package com.pt25.study.example.interfaces.facade.test;

import com.pt25.base.util.HttpUtil;
import com.pt25.base.util.JsonUtil;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import okio.BufferedSource;
import org.jetbrains.annotations.NotNull;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Flux;

import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * http流式数据输出
 */
@Tag(name = "stream http test api")
@RestController
@RequestMapping("/test-stream-response-example")
@Slf4j
public class StreamResponseTestController {


    @GetMapping("/get-items")
    public Object getItems(@RequestParam(required = false) Boolean stream) {
        List<Item> items = Arrays.asList(
                new Item("1", "商品1", 100.0),
                new Item("2", "商品2", 200.0),
                new Item("3", "商品3", 300.0)
        );

        if (Boolean.TRUE.equals(stream)) {
            // 流式返回
            return Flux.fromIterable(items)
                    .delayElements(Duration.ofSeconds(1)) // 每秒输出一个
                    .map(item -> ServerSentEvent.builder()
                            .data(item)
                            .build());
        } else {
            // 普通返回
            return items;
        }
    }

    /**
     * 模拟流式数据输出
     */
    @PostMapping(value = "/get-stream-items", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Object>> getItemsStream(@RequestBody Item itemReq) {

        log.info("get-stream-items params {}", JsonUtil.toJson(itemReq));

        List<Item> items = Arrays.asList(
                new Item("1", "商品1", 100.0),
                new Item("2", "商品2", 200.0),
                new Item("3", "商品3", 300.0),
                new Item("3", "商品4", 300.0)
        );

        return Flux.fromIterable(items)
                .delayElements(Duration.ofSeconds(1))
                .map(item -> ServerSentEvent.builder()
                        .id(item.getId())
                        .event("item-event")
                        .data(item)
                        .build());
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    private static class Item {
        private String id;
        private String name;
        private Double price;
    }

    private final OkHttpClient okHttp = new OkHttpClient.Builder()
            .connectTimeout(10, TimeUnit.SECONDS)
            .readTimeout(0, TimeUnit.SECONDS)   // 无限等待流
            .build();


    /**
     * 基于http的stream流式输出
     */
    @PostMapping(value = "/okhttp-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamChat() {
        SseEmitter emitter = new SseEmitter(60_000L);
        Callback callback = new Callback() {
            @Override
            public void onFailure(@NotNull Call call, @NotNull IOException e) {
                emitter.completeWithError(e);
            }

            @Override
            public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {
                assert response.body() != null;
                try (BufferedSource source = response.body().source()) {
                    String line;
                    while ((line = source.readUtf8Line()) != null) {
                        if (line.startsWith("data:")) {
                            String data = line.substring(5).trim();
                            if ("[DONE]".equals(data)) break;
                            emitter.send(data);   // 逐块推送
                        }
                    }
                    emitter.complete();
                }
            }
        };
        Item item = Item.builder().id("id").name("apple").price(20d).build();
        String url = "http://localhost:6622/test-stream-response-example/get-stream-items";
        Map<String, String> headers = new HashMap<>();
        headers.put(HttpHeaders.CONTENT_TYPE, "application/json");
        HttpUtil.streamPost(url, headers, JsonUtil.toJson(item), callback, 5000, 5000, 5000);
        return emitter;
    }

    /**
     * 基于flux的stream流式输出
     */
    @PostMapping(value = "/flux-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> fluxChat() {
        HttpHeaders headers = new HttpHeaders();
        headers.add("Authorization", "test");
        Item item = Item.builder().id("id").name("apple").price(20d).build();
        return WebClient.create().post()
                .uri("http://localhost:6622/test-stream-response-example/get-stream-items")
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                .bodyValue(item)
                .retrieve()
                .bodyToFlux(String.class)
                .transform(this::processStream) // 流处理
                .onErrorResume(this::handleError);
    }

    /**
     * 流数据处理（转换、过滤、增强等）
     */
    private Flux<String> processStream(Flux<String> originalStream) {
        return originalStream
                .map(data -> {
                    // 这里可以对数据进行处理
                    // 例如：添加时间戳、转换格式等
                    return "{\"processed\": true, \"data\": " + data + ", \"timestamp\": \"" +
                            java.time.Instant.now() + "\"}";
                })
                .filter(data -> !data.contains("error")) // 过滤错误数据
                .delayElements(Duration.ofMillis(50)); // 控制输出频率
    }

    /**
     * 错误处理
     */
    private Flux<String> handleError(Throwable error) {
        return Flux.just("{\"error\": \"" + error.getMessage() + "\", \"type\": \"proxy_error\"}")
                .delayElements(Duration.ofSeconds(1));
    }
}
